package com.just.springbootbatch.batch;

import com.just.springbootbatch.domain.Person;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.batch.core.launch.support.RunIdIncrementer;
import org.springframework.batch.core.launch.support.SimpleJobLauncher;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.repository.support.JobRepositoryFactoryBean;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.database.BeanPropertyItemSqlParameterSourceProvider;
import org.springframework.batch.item.database.JdbcBatchItemWriter;
import org.springframework.batch.item.file.FlatFileItemReader;
import org.springframework.batch.item.file.mapping.BeanWrapperFieldSetMapper;
import org.springframework.batch.item.file.mapping.DefaultLineMapper;
import org.springframework.batch.item.file.transform.DelimitedLineTokenizer;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.ClassPathResource;
import org.springframework.transaction.PlatformTransactionManager;

import javax.sql.DataSource;

/**
 * 配置batch，手动触发
 */
@Configuration
@EnableBatchProcessing
public class TriggerBatchConfig {

    /**
     * 读取数据
     * 定义itemReader
     */
    @Bean(name="triggerReader")
    @StepScope
    public FlatFileItemReader<Person> reader(@Value("#{jobParameters['input.file.name']}") String pathToFile) throws Exception {
        FlatFileItemReader<Person> reader = new FlatFileItemReader<Person>(); //1
        reader.setResource(new ClassPathResource(pathToFile)); //2
        reader.setLineMapper(new DefaultLineMapper<Person>() {{ //3
            setLineTokenizer(new DelimitedLineTokenizer() {{
                setNames(new String[]{"name", "age", "nation", "address"});
            }});
            setFieldSetMapper(new BeanWrapperFieldSetMapper<Person>() {{
                setTargetType(Person.class);
            }});
        }});

        return reader;
    }

    /**
     * 处理数据
     * 定义自己的ItemProcessor
     */
    @Bean(name="triggerProcessor")
    public ItemProcessor<Person, Person> processor() {
        CsvItemProcessor csvItemProcessor = new CsvItemProcessor();
        csvItemProcessor.setValidator(csvBeanValidator());        //指定校验器为自己的csvBeanValidator
        return csvItemProcessor;
    }

    @Bean
    public CsvBeanValidator<Person> csvBeanValidator() {
        return new CsvBeanValidator<>();
    }

    /**
     * 输出数据
     * 定义ItemWriter
     * 用JDBC批处理JdbcBatchItemWriter将数据写入数据库
     * 注意我此处用的表是batch_person
     */
    @Bean(name="triggerWriter")
    public ItemWriter<Person> writer(DataSource dataSource) {
        JdbcBatchItemWriter<Person> writer = new JdbcBatchItemWriter<>();
        writer.setItemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider<>());
        String sql = "insert into batch_person(name,age,nation,address)"
                + "values(:name,:age,:nation,:address)";
        writer.setSql(sql);
        writer.setDataSource(dataSource);
        return writer;
    }

    /**
     * 用来注册job的容器
     * 定义JobRepository
     * springboot已经自动配置了dataSource和transactionManager,可以通过方法直接注入这两个bean
     */
    @Bean
    public JobRepository jobRepository(DataSource dataSource, PlatformTransactionManager platformTransactionManager) throws Exception {
        JobRepositoryFactoryBean jobRepositoryFactoryBean = new JobRepositoryFactoryBean();
        jobRepositoryFactoryBean.setDatabaseType("mysql");
        jobRepositoryFactoryBean.setDataSource(dataSource);
        jobRepositoryFactoryBean.setTransactionManager(platformTransactionManager);
        return jobRepositoryFactoryBean.getObject();
    }

    /**
     * 用来启动job的接口
     * 定义jobLauncher
     */
    @Bean
    public SimpleJobLauncher jobLauncher(DataSource dataSource, PlatformTransactionManager platformTransactionManager) throws Exception {
        SimpleJobLauncher simpleJobLauncher = new SimpleJobLauncher();
        simpleJobLauncher.setJobRepository(jobRepository(dataSource, platformTransactionManager));
        return simpleJobLauncher;
    }

    /**
     * 需要实际执行的job
     * 定义job
     */
    @Bean(name = "triggerJob")
    public Job importJob(JobBuilderFactory jobs, @Qualifier("triggerStep1") Step s1) {
        return jobs.get("importJob")
                .incrementer(new RunIdIncrementer())
                .flow(s1)                         //指定step
                .end()
                .listener(csvJobListener())       //绑定监听器
                .build();
    }

    /**
     * 步骤，包含ItemReader，ItemProcessor，ItemReader
     */
    @Bean(name = "triggerStep1")
    public Step step1(StepBuilderFactory stepBuilderFactory,
                      @Qualifier("triggerReader") ItemReader<Person> reader,
                      @Qualifier("triggerWriter")ItemWriter<Person> writer,
                      @Qualifier("triggerProcessor")ItemProcessor<Person, Person> processor) {
        return stepBuilderFactory.get("step1")
                .<Person, Person>chunk(10)    //每次提交n条数据
                .reader(reader)                            //step绑定reader
                .processor(processor)                      //step绑定processor
                .writer(writer)                            //step绑定writer
                .build();
    }

    @Bean
    public CsvJobListener csvJobListener() {
        return new CsvJobListener();
    }
}
